#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/fcntl.h>
#include <sys/mman.h>
#include <unistd.h>
#include <string.h>
#include <cuda.h>
#include "scanImpl.cu"

#define BI 1000000000
#define HSIZE (1024*1024*10)

#  define CUDA_SAFE_CALL_NO_SYNC( call) do {                                \
    cudaError err = call;                                                    \
    if( cudaSuccess != err) {                                                \
        fprintf(stderr, "Cuda error in file '%s' in line %i : %s.\n",        \
                __FILE__, __LINE__, cudaGetErrorString( err) );              \
        exit(EXIT_FAILURE);                                                  \
    } } while (0)
#  define CUDA_SAFE_CALL( call) do {                                        \
    CUDA_SAFE_CALL_NO_SYNC(call);                                            \
    cudaError err = cudaThreadSynchronize();                                 \
    if( cudaSuccess != err) {                                                \
        fprintf(stderr, "Cuda error in file '%s' in line %i : %s.\n",        \
                __FILE__, __LINE__, cudaGetErrorString( err) );              \
        exit(EXIT_FAILURE);                                                  \
    } } while (0)


__global__ void count_hash_num(char * dim, int dNum,int *num){
	int stride = blockDim.x * gridDim.x;
	int offset = blockIdx.x * blockDim.x + threadIdx.x;

	for(int i=offset;i<dNum;i+=stride){
		int tmp = ((int *)dim)[i];
		int hKey = tmp %HSIZE;
		atomicAdd(&(num[hKey]),1);
	}

}

__global__ void build_hash_table(char *dim, int dNum, int *psum, char * bucket){

	int stride = blockDim.x * gridDim.x;
	int offset = blockIdx.x * blockDim.x + threadIdx.x;

	for(int i=offset;i<dNum;i+=stride){
		int joinKey = ((int *) dim)[i];
                int hKey = joinKey % HSIZE;
                int offset = atomicAdd(&psum[hKey],1) * 2;
		((int*)bucket)[offset] = joinKey;
                offset += 1;
                int dimId = i+1;
                ((int*)bucket)[offset] = dimId;
	}
}


__global__ void count_join_result(int * num, int * psum, char * bucket, char * fact, int fNum, int * count, int*factFilter){
	int lcount = 0;
	int startIndex;
	int stride = blockDim.x * gridDim.x;
	int offset = blockIdx.x*blockDim.x + threadIdx.x; 

	startIndex = blockIdx.x*blockDim.x + threadIdx.x;
	for(int i=startIndex;i<fNum;i+=stride){
		int fkey = ((int *)(fact))[i];
                int hkey = fkey % HSIZE;
                int keyNum = num[hkey];
		int fvalue = 0;

                for(int j=0;j<keyNum;j++){
                	int pSum = psum[hkey];
                        int dimKey = ((int *)(bucket))[2*j + 2*pSum];
                        int dimId = ((int *)(bucket))[2*j + 2*pSum + 1];
                        if( dimKey == fkey){
                                lcount ++;
				fvalue = dimId;
                                break;
                        }
               	} 
                factFilter[i] = fvalue;

	}
	count[offset] = lcount;	
}

__global__ void static joinFact(int *resPsum, char * fact,  int attrSize, long  num, int * filter, char * result){

        int startIndex = blockIdx.x*blockDim.x + threadIdx.x;
        int stride = blockDim.x * gridDim.x;
        long localOffset = resPsum[startIndex];

        for(long i=startIndex;i<num;i+=stride){
                if(filter[i] != 0){
                        ((int *)result)[localOffset] = ((int*)fact)[i];
                        localOffset += 1;
                }
        }

}

__global__ void static joinDim(int *resPsum, char * dim, int attrSize, long num,int * filter, char * result){

        int startIndex = blockIdx.x*blockDim.x + threadIdx.x;
        int stride = blockDim.x * gridDim.x;
        long localOffset = resPsum[startIndex];

        for(long i=startIndex;i<num;i+=stride){
                int dimId = filter[i];
                if( dimId != 0){
                        ((int *)result)[localOffset]  = ((int*)dim)[dimId-1];
                        localOffset += 1;
                }
        }

}


int main(int argc, char * argv[]){
	int *fact, *dim;
	int dimSize, factSize;
	int factNum, dimNum;

	int *cpu_count, *resPsum;
	int count = 0;

	int * gpu_hashNum;
	char * gpu_result;
	char *gpu_bucket, *gpu_fact, * gpu_dim;
	int * gpu_count,  *gpu_psum, *gpu_resPsum;

	struct timespec start,end;
	float gpuTime;
	char **factTable, **dimTable;

	if(argc!=5){
		printf("error: factTable dimTable fColNum dColNum\n");
		exit(-1);
	}

	int * tmp;
	CUDA_SAFE_CALL_NO_SYNC(cudaMalloc((void **)&tmp, sizeof(int)));
	CUDA_SAFE_CALL_NO_SYNC(cudaFree(tmp));

	int fColNum = atoi(argv[3]);
	int dColNum = atoi(argv[4]);

	fact =(int *) malloc(sizeof(int) * fColNum);
	dim = (int *) malloc(sizeof(int) * dColNum);
	dimTable = (char **) malloc(sizeof(char *) * dColNum);
	factTable = (char **) malloc(sizeof(char *) * fColNum);

	if(!fact || !dim ){
		printf("Failed to allocate host memory\n");
		exit(-1);
	}

	for(int i=0;i<fColNum;i++){
		char buf[32] = {0};
		sprintf(buf,"%s%d",argv[1],i);
		fact[i] = open(buf,O_RDONLY);
	}
	for(int i = 0;i<dColNum;i++){
		char buf[32] = {0};
		sprintf(buf,"%s%d",argv[2],i);
		dim[i] = open(buf, O_RDONLY);
	}

	dimSize = lseek(dim[0],0,SEEK_END);
	factSize = lseek(fact[0],0,SEEK_END);
	factNum = factSize/sizeof(int);
	dimNum = dimSize/sizeof(int);

	for(int i=0;i<fColNum;i++){
		//factTable[i] = (char *) malloc(factSize);
		cudaMallocHost((void**)&factTable[i],factSize);
		char *factT = (char *)mmap(0,factSize,PROT_READ,MAP_SHARED,fact[i],0);
		memcpy(factTable[i],factT,factSize);
		munmap(factT,factSize);
	}
	for(int i=0;i<dColNum;i++){
		//dimTable[i] = (char *) malloc(dimSize);
		cudaMallocHost((void**)&dimTable[i],dimSize);
		char *dimT = (char *)mmap(0,dimSize,PROT_READ,MAP_SHARED,dim[i],0);
		memcpy(dimTable[i],dimT,dimSize);
		munmap(dimT,dimSize);
	}

	int * gpuTmp;
	CUDA_SAFE_CALL_NO_SYNC(cudaMalloc((void**)&gpuTmp, sizeof(int)));
	CUDA_SAFE_CALL_NO_SYNC(cudaFree(gpuTmp));

	dim3 grid(2048);
	dim3 block(256);
	int threadNum = grid.x * block.x;
	cpu_count = (int *) malloc(sizeof(int)*threadNum);
	memset(cpu_count,0,sizeof(int)*threadNum);
	resPsum = (int *) malloc(sizeof(int)*threadNum);
	memset(resPsum,0,sizeof(int)*threadNum);


/*
 * 	build hash table on GPU
 */
	
	clock_gettime(CLOCK_REALTIME,&start);

	cudaEvent_t startGPU, stopGPU;
	cudaEventCreate(&startGPU);
	cudaEventCreate(&stopGPU);
	cudaEventRecord(startGPU,0);
	int *gpu_psum1;

	CUDA_SAFE_CALL(cudaMalloc((void**)&gpu_dim,dimSize));
	CUDA_SAFE_CALL(cudaMemcpy(gpu_dim,dimTable[0],dimSize,cudaMemcpyHostToDevice));
	CUDA_SAFE_CALL(cudaMalloc((void**)&gpu_hashNum,sizeof(int)*HSIZE));
	CUDA_SAFE_CALL(cudaMemset(gpu_hashNum,0,sizeof(int)*HSIZE));

	count_hash_num<<<grid,block>>>(gpu_dim,dimNum,gpu_hashNum);
	cudaDeviceSynchronize();

	CUDA_SAFE_CALL(cudaMalloc((void**)&gpu_psum,HSIZE*sizeof(int)));
	CUDA_SAFE_CALL(cudaMalloc((void**)&gpu_psum1,HSIZE*sizeof(int)));

	scanImpl(gpu_hashNum,HSIZE,gpu_psum);

	CUDA_SAFE_CALL(cudaMemcpy(gpu_psum1,gpu_psum,sizeof(int)*HSIZE,cudaMemcpyDeviceToDevice));

	CUDA_SAFE_CALL(cudaMalloc((void **)&gpu_bucket,2*dimSize));

	build_hash_table<<<grid,block>>>(gpu_dim,dimNum,gpu_psum1,gpu_bucket);

	CUDA_SAFE_CALL(cudaFree(gpu_psum1));

/*
 *	join on GPU
*/

	int * gpuFilter;

	CUDA_SAFE_CALL_NO_SYNC(cudaMalloc((void **)&gpuFilter,factSize));
	CUDA_SAFE_CALL_NO_SYNC(cudaMalloc((void**)&gpu_fact,factSize));
	count = HSIZE;


	CUDA_SAFE_CALL_NO_SYNC(cudaMemcpy(gpu_fact,factTable[0],factSize,cudaMemcpyHostToDevice));
	CUDA_SAFE_CALL_NO_SYNC(cudaMalloc((void**)&gpu_count,sizeof(int)*threadNum));

	count_join_result<<<grid,block>>>(gpu_hashNum, gpu_psum, gpu_bucket, gpu_fact, factNum,gpu_count,gpuFilter);

	CUDA_SAFE_CALL_NO_SYNC(cudaMemcpy(cpu_count,gpu_count,sizeof(int)*threadNum,cudaMemcpyDeviceToHost));
	CUDA_SAFE_CALL_NO_SYNC(cudaMalloc((void**)&gpu_resPsum, sizeof(int)*threadNum));

	scanImpl(gpu_count,threadNum,gpu_resPsum);

	CUDA_SAFE_CALL_NO_SYNC(cudaMemcpy(resPsum,gpu_resPsum,sizeof(int)*threadNum,cudaMemcpyDeviceToHost));

	count = resPsum[threadNum-1] + cpu_count[threadNum-1];

	CUDA_SAFE_CALL_NO_SYNC(cudaMalloc((void**)&gpu_result,count * sizeof(int)));

	for(int i=1;i<fColNum;i++){
		CUDA_SAFE_CALL_NO_SYNC(cudaMemcpy(gpu_fact,factTable[i],factSize,cudaMemcpyHostToDevice));
		joinFact<<<grid,block>>>(gpu_resPsum,gpu_fact, sizeof(int), factNum,gpuFilter,gpu_result);

		CUDA_SAFE_CALL_NO_SYNC(cudaMemcpy(factTable[i],(void*)gpu_result,count*sizeof(int),cudaMemcpyDeviceToHost));
	}

	for(int i=1; i<dColNum;i++){
		CUDA_SAFE_CALL_NO_SYNC(cudaMemcpy(gpu_dim,dimTable[i],dimSize,cudaMemcpyHostToDevice));
		joinDim<<<grid,block>>>(gpu_resPsum,gpu_dim, sizeof(int), factNum, gpuFilter,gpu_result);
		CUDA_SAFE_CALL_NO_SYNC(cudaMemcpy(factTable[i],(void*)gpu_result,count*sizeof(int),cudaMemcpyDeviceToHost));
	}

	CUDA_SAFE_CALL(cudaFree(gpu_fact));
	CUDA_SAFE_CALL(cudaFree(gpu_dim));
	CUDA_SAFE_CALL(cudaFree(gpu_hashNum));
	CUDA_SAFE_CALL(cudaFree(gpu_count));
	CUDA_SAFE_CALL(cudaFree(gpu_psum));
	CUDA_SAFE_CALL(cudaFree(gpu_bucket));
	CUDA_SAFE_CALL(cudaFree(gpu_result));

	for(int i=0;i<fColNum;i++){
		close(fact[i]);
		cudaFreeHost(factTable[i]);
	}
	for(int i = 0;i<dColNum;i++){
		close(dim[i]);
		cudaFreeHost(dimTable[i]);
	}

	cudaEventRecord(stopGPU,0);
	cudaEventSynchronize(stopGPU);
	cudaEventElapsedTime(&gpuTime,startGPU,stopGPU);
	printf("GPU total time:%lf\n",gpuTime);

	clock_gettime(CLOCK_REALTIME,&end);
	double timeE = (end.tv_sec -  start.tv_sec)* BI + end.tv_nsec - start.tv_nsec;
	printf("CPU time: %lf\n", timeE);
	return 0;
}
